package com.sui.bigdata.sml.web.service.impl;

import com.feidee.fdcommon.configuration.CustomConfiguration;
import com.sui.bigdata.sml.web.dto.ModelDTO;
import com.sui.bigdata.sml.web.dto.ModelGroupDTO;
import com.sui.bigdata.sml.web.dto.ModelGroupInfoDTO;
import com.sui.bigdata.sml.web.exception.EntityNotFoundException;
import com.sui.bigdata.sml.web.exception.InactiveModelException;
import com.sui.bigdata.sml.web.exception.InvalidTokenException;
import com.sui.bigdata.sml.web.exception.StartModelServingException;
import com.sui.bigdata.sml.web.misc.k8s.KubeClient;
import com.sui.bigdata.sml.web.misc.k8s.KubeComponentMaker;
import com.sui.bigdata.sml.web.misc.k8s.ServingInfo;
import com.sui.bigdata.sml.web.misc.k8s.ServingModel;
import com.sui.bigdata.sml.web.repository.mapper.ModelGroupMapper;
import com.sui.bigdata.sml.web.repository.mapper.ModelMapper;
import com.sui.bigdata.sml.web.repository.model.Model;
import com.sui.bigdata.sml.web.repository.model.ModelGroup;
import com.sui.bigdata.sml.web.service.ModelGroupService;
import com.sui.bigdata.sml.web.util.Constants;
import com.sui.bigdata.sml.web.util.ModelType;
import com.sui.bigdata.sml.web.util.Tools;
import io.kubernetes.client.ApiException;
import io.kubernetes.client.models.V1Deployment;
import lombok.extern.slf4j.Slf4j;
import ma.glasnost.orika.MapperFacade;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.DependsOn;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @author songhaicheng
 * @date 2020/1/15 10:05
 * @description
 * @reviewer
 */
@Service
@DependsOn("disconfConfig")
@Slf4j
public class ModelGroupServiceImpl implements ModelGroupService {

    @Autowired
    private ModelGroupMapper mapper;
    @Autowired
    private ModelMapper modelMapper;
    @Autowired
    private MapperFacade mapperFacade;

    @Autowired
    private KubeClient kubeClient;

    @Transactional
    @Override
    public ModelGroupDTO add(ModelGroupDTO dto) {
        checkModels(dto.getModels());
        String tokens = String.join(",", dto.getModels());
        ModelGroup group = new ModelGroup()
                .setModelTokens(tokens)
                .setServingToken(Tools.MD5(tokens).substring(8, 24))
                .setName(dto.getName())
                .setDescription(dto.getDescription())
                .setUserId(Constants.DEFAULT_USER_ID)
                .setActive(true);
        if (mapper.getByServingToken(group.getServingToken()) != null) {
            throw new RuntimeException("模型组已存在：" + group.getServingToken() + " -> " + tokens);
        }
        // 先插入数据库，再进行耗时较长的 K8S 启动组件动作，防止此时有其他人又做了遍新上线动作
        if (mapper.insert(group) != 1) {
            throw new RuntimeException("保存模型组失败：" + group.getModelTokens());
        }
        createKubeEntity(group);
        return mapperFacade.map(group, ModelGroupDTO.class);
    }

    @Override
    public ModelGroupDTO get(Long id) {
        ModelGroup group = mapper.get(id);
        if (group == null) {
            throw new EntityNotFoundException(ModelGroup.class, id);
        }
        return mapperFacade.map(group, ModelGroupDTO.class);
    }

    @Override
    public ModelGroupDTO getByServingToken(String servingToken) {
        ModelGroup group = mapper.getByServingToken(servingToken);
        if (group == null) {
            throw new InvalidTokenException(servingToken);
        }
        return mapperFacade.map(group, ModelGroupDTO.class);
    }

    @Override
    public ModelGroupInfoDTO getInfoByServingToken(String servingToken) {
        ModelGroup group = mapper.getByServingToken(servingToken);
        if (group == null) {
            throw new InvalidTokenException(servingToken);
        }
        ModelGroupInfoDTO dto = mapperFacade.map(group, ModelGroupInfoDTO.class);
        // 获取里面模型的具体信息
        List<Model> models = listModelsByModelStrWithDelimiter(group.getModelTokens(), ",");
        dto.setModels(mapperFacade.mapAsList(models, ModelDTO.class));
        if (dto.getActive()) {
            dto.setUrl(CustomConfiguration.getString(Constants.SERVING_URL_PREFIX, "https://tg.feidee.net/model/") + dto.getToken());
        } else {
            dto.setUrl("未上线");
        }
        return dto;
    }

    @Transactional
    @Override
    public Boolean modifyServingActive(String servingToken, boolean active) {
        ModelGroup group = mapperFacade.map(getByServingToken(servingToken), ModelGroup.class);
        // 先改变数据库的状态，防止中途又有其他人做了同样的操作
        boolean activeBefore = group.getActive();
        group.setActive(active);
        boolean dbUpdated = mapper.update(group) == 1;
        if (activeBefore != active) {
            if (active) {
                createKubeEntity(group);
            } else {
                deleteKubeEntity(group);
            }
        }
        return dbUpdated;
    }

    @Override
    public Boolean modifyServingModels(String servingToken, List<String> modelTokens) {
        checkModels(modelTokens);
        ModelGroup group = mapperFacade.map(getByServingToken(servingToken).setModels(modelTokens), ModelGroup.class);
        group.setActive(null);
        return mapper.update(group) == 1;
    }

    @Override
    public List<ModelGroupDTO> listByUserId(Long userId) {
        return mapperFacade.mapAsList(mapper.listByUserId(userId), ModelGroupDTO.class);
    }

    @Override
    public List<String> listServingInstances(String servingToken) {
        try {
            return kubeClient.listInstance(servingToken);
        } catch (ApiException e) {
            log.error("error: ", e);
            throw new RuntimeException("查看模型服务示例失败：" + servingToken);
        }
    }

    @Override
    public List<ModelGroupDTO> listHavingModelToken(String modelToken) {
        return mapperFacade.mapAsList(mapper.listHavingModelToken(modelToken, true), ModelGroupDTO.class);
    }

    @Override
    public Boolean refreshServing(String servingToken) {
        ModelGroup group = mapperFacade.map(getByServingToken(servingToken), ModelGroup.class);
        V1Deployment result = replaceKubeEntity(group);
        if (result.getStatus().getAvailableReplicas() > 0) {
            log.warn("replace deployment success：{}", servingToken);
        } else {
            log.error("replace deployment failure：{}", servingToken);
            throw new StartModelServingException(group.getServingToken());
        }
        return true;
    }

    public List<Model> listModelsByModelStrWithDelimiter(String modelStr, String delimiter) {
        List<Model> models = new ArrayList<>();
        for (String modelToken : modelStr.split(delimiter)) {
            List<Model> activeModels = modelMapper.listByTokenWithVersionAndOnline(modelToken, null, true);
            if (activeModels == null || activeModels.size() == 0) {
                throw new InactiveModelException(modelToken);
            }
            models.add(activeModels.get(0));
        }
        return models;
    }

    /**
     * 确保加入模型组的模型都是合法的（全部激活 & 只有一个版本被激活 & 最多一个 tf 模型）
     *
     * @param modelTokens
     */
    public void checkModels(List<String> modelTokens) {
        int tfCnt = 0;
        for (String token : modelTokens) {
            List<Model> models = modelMapper.listByTokenWithVersionAndOnline(token, null, true);
            if (models.size() == 0) {
                throw new RuntimeException("模型未激活：" + token);
            } else if (models.size() > 1) {
                // 模型多个版本被激活，只保留版本号最大的那个版本
                models.parallelStream().forEach(model -> model.setOnline(false));
                models.get(0).setOnline(true);
                modelMapper.updateBatch(models);
                throw new RuntimeException("检测到模型有多个激活版本，已初始化激活唯一版本 "
                        + models.get(0).getVersion() + "：" + token + "\n请确认后重新提交该次请求");
            }
            // 最多只能有一个 tf 模型
            if (ModelType.getType(models.get(0).getModelTypes()) == ModelType.TF) {
                tfCnt++;
            }
            if (tfCnt > 1) {
                throw new RuntimeException("只允许出现至多一个 tf 模型");
            }
        }
    }

    private void createKubeEntity(ModelGroup group) {
        List<Model> models = listModelsByModelStrWithDelimiter(group.getModelTokens(), ",");
        ServingInfo servingInfo = new ServingInfo(group.getServingToken(), mapperFacade.mapAsList(models, ServingModel.class));
        KubeComponentMaker maker = new KubeComponentMaker(servingInfo);
        try {
            kubeClient.createDeployment(maker.getDeployment());
            kubeClient.createService(maker.getService());
            kubeClient.createIngress(maker.getIngress());
            if (!kubeClient.testEntity(group.getServingToken())) {
                throw new Exception("testEntity error");
            }
        } catch (Exception e) {
            log.error("error: ", e);
            deleteKubeEntity(group);
            throw new StartModelServingException(group.getServingToken());
        }
    }

    private void deleteKubeEntity(ModelGroup group) {
        try {
            kubeClient.deleteDeployment(group.getServingToken());
            kubeClient.deleteService(group.getServingToken());
            kubeClient.deleteIngress(group.getServingToken());
        } catch (ApiException e) {
            log.error("error: ", e);
            throw new RuntimeException("删除模型预测服务失败：" + group.getServingToken());
        }
    }

    private V1Deployment replaceKubeEntity(ModelGroup group) {
        List<Model> models = Arrays.stream(group.getModelTokens().split(",")).map(modelToken -> {
            List<Model> activeModels = modelMapper.listByTokenWithVersionAndOnline(modelToken, null, true);
            if (activeModels == null || activeModels.size() == 0) {
                throw new InactiveModelException(modelToken);
            }
            return activeModels.get(0);
        }).collect(Collectors.toList());
        ServingInfo info = new ServingInfo(group.getServingToken(), mapperFacade.mapAsList(models, ServingModel.class));
        KubeComponentMaker maker = new KubeComponentMaker(info);

        try {
            return kubeClient.replaceDeployment(group.getServingToken(), maker.getDeployment());
        } catch (ApiException e) {
            log.error("error: ", e);
            throw new StartModelServingException(group.getServingToken());
        }
    }
}
